﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Reflection;
using System.Runtime.Serialization;
using System.Threading.Actors.Languages;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    public abstract class ActorClient : IObserver<ActorEvent>, IDisposable
    {
        // The actor proxy
#pragma warning disable 3026  // protected volatile fields aren't CLS compliant
        protected volatile ActorRawTcpClient m_actorTcpClient; // May be accessed by multiple threads, and will be mutable
#pragma warning restore 3026

        // State items that must be cached in order to re-resolve the actor
        protected readonly bool m_useGateway; 
        protected readonly Uri m_gatewayUri;
        protected readonly Uri m_serverUri;
        protected readonly Uri m_actorUri;
        
        // Tracking IObservers instead of ISubjects because there should be at most one IObserver per eventType.
        protected Dictionary<string, IObserver<ActorEvent>> m_observers = new Dictionary<string,IObserver<ActorEvent>>();

        // Protects any access to m_actorTcpClient.  Necessary because we may get asynchronous fault notifications from the channel.
        protected readonly object m_operationLockObject = new object();

        // If we are using partitioning, which service partition do we connect to?  Null is accepted for singleton partitions (the default).
        protected readonly String m_partitionName;

        // An implementation-specific part of a URI for locating an actor in the cluster.
        protected String m_actorDiscriminator;

        // Assists in converting exception type strings to exceptions
        private static readonly Dictionary<string, Func<string, Exception>> s_knownExceptionTable;

        // The Guid should be unique across this machine, guaranteeing that we won't run into conflicts in other
        // appdomains or processes.
        private static readonly string baseId = Guid.NewGuid().ToString();
        private static int identificationIndex = 0;
        protected static string GetUniqueId()
        {
            int index = Interlocked.Increment(ref identificationIndex);
            return baseId + "-" + index;
        }

        protected readonly string myId;
        private int mySequenceNumber = 0;
        protected int GetNewSequenceNumber() { return Interlocked.Increment(ref mySequenceNumber); }

        // Retry values
        // A service instance may take easily 16 seconds from the time it gets load balanced to when it is available again,
        // under moderate load.  Let's make sure these values mean we will wait for 30 seconds.
        protected static readonly int CONNECT_RETRIES = 30;
        protected static readonly int CONNECT_RETRY_WAIT_TIME_MS = 1000;
        protected static readonly int COMMAND_RETRIES = 5;
        // Ensure we wait at least 15 seconds for a modest failover scenario.
        private const int TestConnectionTimeoutMS = 80 * 1000;
        private const int MinConnectionTimeoutMS = 15 * 1000;
        private const int MinDebuggerAttachedConnectionTimeoutMS = 10 * 60 * 1000;

        // Keep the resolve state - 0 unresolved, 1 resolving, 2 resolved
        protected ResolverState m_resolveState;

        // Custom resolver for obtaining endpoint Uri from Fabric and Actor
        protected readonly Func<Uri, Uri, Uri> m_resolver;

        // Record the fact that we've been disposed
#pragma warning disable 3026  // protected volatile fields aren't CLS compliant
        protected volatile bool m_disposed = false;
#pragma warning restore 3026

        // Record whether we've asked the actor to delete itself already.  If so, no need to send it more messages.
        protected bool m_sentDeleteRequest = false;

        // Actor event processing uses delimiters for separating actor names from event names, 
		// partition information, etc.
        public static readonly char[] ReservedDelimiters = new char[] { '$', '#' };

        // Support for factory methods that use real implementations of ActorClient.
        private const String ActorAssemblyName = "System.Threading.Actors.WinFab, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null";
        private const String FabricActorClientTypeName = "System.Threading.Actors.FabricActorClient";
        private static Type s_clientType;
        private static Func<String, string, Uri, String, bool, int, int, string, ActorClient> s_createOrConnectMethod;

        //
        // Static constructor
        //

        static ActorClient()
        {
            // Set up our known exceptions
            s_knownExceptionTable = new Dictionary<string,Func<string,Exception>>();
            // .NET Framework exceptions
            s_knownExceptionTable.Add("ArgumentException", (arg) => new ArgumentException(arg));
            s_knownExceptionTable.Add("ArgumentNullException", (arg) => new ArgumentNullException("unknown", arg));
            s_knownExceptionTable.Add("ArgumentOutOfRangeException", (arg) => new ArgumentOutOfRangeException("unknown", arg));
            s_knownExceptionTable.Add("DivideByZeroException", (arg) => new DivideByZeroException(arg));  // for testing more than anything
            s_knownExceptionTable.Add("IndexOutOfRangeException", (arg) => new IndexOutOfRangeException(arg));
            s_knownExceptionTable.Add("InvalidOperationException", (arg) => new InvalidOperationException(arg));
            s_knownExceptionTable.Add("NotImplementedException", (arg) => new NotImplementedException(arg));
            s_knownExceptionTable.Add("NotSupportedException", (arg) => new NotSupportedException(arg));
            s_knownExceptionTable.Add("NullReferenceException", (arg) => new NullReferenceException(arg));
            s_knownExceptionTable.Add("OperationCanceledException", (arg) => new OperationCanceledException(arg));  // Saw this in stress on Azure.  Resolution timeouts?
            s_knownExceptionTable.Add("SerializationException", (arg) => new SerializationException(arg));
            s_knownExceptionTable.Add("TimeoutException", (arg) => new TimeoutException(arg));

            // Actor Framework specific exceptions
            s_knownExceptionTable.Add("ActorNotFoundException", (arg) => new ActorNotFoundException(arg));
        }

        //
        // Constructor
        //

        // Derived classes should call SetupActorClient after 
        public ActorClient(Uri fabricUri, Uri actorUri, bool useGateway, String partitionName = null, Func<Uri, Uri, Uri> resolver = null)
        {
            Contract.Requires(actorUri != null);
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorUri.ToString().Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");

            // These values need to be cached (in addition to m_gatewayUri) so that they will be
            // available in the event that we need to re-resolve the actor.
            m_actorUri = actorUri;
            m_serverUri = fabricUri;
            m_useGateway = useGateway;
            m_resolver = resolver;
            m_partitionName = partitionName;

            if (useGateway)
            {
                m_gatewayUri = MakeGatewayAddress(fabricUri);
            }

            myId = GetUniqueId();
            
            m_resolveState = ResolverState.Unresolved;
        }

        ~ActorClient()
        {
            Dispose(false);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!m_disposed)
            {
                m_disposed = true;

                foreach (var observer in m_observers.Values)
                    observer.OnCompleted();
                m_observers.Clear();

                var localActorTcpClient = m_actorTcpClient;
                if (localActorTcpClient != null)
                {
                    if (!m_sentDeleteRequest)
                        UnsubscribeAllFireAndForget();
                    localActorTcpClient.Faulted -= new EventHandler(ActorClient_Faulted);
                    localActorTcpClient.Dispose();
                }
                m_actorTcpClient = null;
            }
        }

        //
        // Factory methods
        //

        private static void FindActorClientImplementation()
        {
            if (s_clientType == null)
            {
                // Try loading known implementations.
                s_clientType = Type.GetType(FabricActorClientTypeName + ", " + ActorAssemblyName, false);

                if (s_clientType == null)
                    s_clientType = typeof(GatewayActorClient);
            }
            Contract.Assert(s_clientType != null, "Could not find an implementation subclass for ActorClient.  Ensure that FabricActorClient or another implementation is already loaded.");

            if (s_clientType != null && s_createOrConnectMethod == null)
            {
                var method = s_clientType.GetRuntimeMethod("CreateOrConnectToActor", 
                    new Type[] { typeof(String), typeof(String), typeof(Uri), typeof(String), typeof(Boolean), 
                                 typeof(int), typeof(int), typeof(String) });
                Contract.Assert(method != null, "Couldn't find implementation of CreateOrConnectToActor");
                s_createOrConnectMethod = (Func<String, string, Uri, String, bool, int, int, string, ActorClient>)
                    method.CreateDelegate(typeof(Func<String, string, Uri, String, bool, int, int, string, ActorClient>));
            }
        }

        public static ActorClient CreateOrConnectToActor(string fabricAddress, string applicationType, Uri actorUri,
            string actorTypeName, bool connectThroughGateway, int numReplicas = 2,
            int numPartitions = 1, String partitionName = null)
        {
            if (s_createOrConnectMethod == null)
                FindActorClientImplementation();
            ActorClient client = s_createOrConnectMethod(fabricAddress, applicationType, 
                actorUri, actorTypeName, connectThroughGateway, numReplicas, numPartitions, partitionName);
            return client;
        }

        #region Public helpers

        public Uri ActorUri
        {
            get { return m_actorUri; }
        }

        public ResolverState ResolvedState
        {
            get { return m_resolveState; }
        }
		
        public Uri ServerUri {
            get { return m_serverUri; }
        }

        public abstract void DeleteActor();

        public String PartitionName {
            get { return m_partitionName; }
        }

        #endregion Public helpers

        //
        // Wrappers for IActor methods
        //

        #region JavaScript actor methods

        public bool AddMethod(SupportedLanguages language, string methodName, string methodImpl)
        {
            return AddMethodAsync((int)language, methodName, methodImpl).Result;
        }

        private Task<bool> AddMethodAsync(int language, string methodName, string methodImpl)
        {
            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.ADDMETHOD, myId, seqNo, language, methodName, methodImpl);
            var completionTask = SendActorRequestAsync(asreq);
            return completionTask.ContinueWith(antecedent =>
                {
                    var asres = antecedent.Result;
                    if (asres.Status == ActorServiceResultStatus.Succeeded)
                    {
                        return asres.GetTypedResult<bool>();
                    }
                    else throw ConstructException(asres);
                }, TaskContinuationOptions.ExecuteSynchronously);
        }

        #endregion JavaScript actor methods

        #region Assembly Deployment

        public bool DoesAssemblyExist(string assemblyName, bool allowHigherVersions)
        {
            // @TODO: This was commented out & returning false before our refactoring....  revisit this one day.
            //return m_actorClient.DoesAssemblyExist(assemblyName, allowHigherVersions);
            return false;
        }

        public bool AddAssembly(string assemblyName, byte[] assemblyBytes)
        {
            return AddAssemblyAsync(assemblyName, assemblyBytes).Result;
        }

        private Task<bool> AddAssemblyAsync(string assemblyName, byte[] assemblyBytes)
        {
            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.ADDASSEMBLY, myId, seqNo, assemblyName, assemblyBytes);
            var completionTask = SendActorRequestAsync(asreq);
            return completionTask.ContinueWith(antecedent =>
                {
                    var asres = antecedent.Result; // will throw on communication failure
                    if (asres.Status == ActorServiceResultStatus.Succeeded)
                    {
                        return asres.GetTypedResult<bool>();
                    }
                    else throw ConstructException(asres);
                }, TaskContinuationOptions.ExecuteSynchronously);
        }

        #endregion Assembly Deployment


        #region CallMethod variants

        public bool CallMethodNoThrow<T>(string methodName, object[] parameters, out T result, out Exception thrownException)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);
            Contract.Ensures((Contract.Result<bool>() && Contract.ValueAtReturn<Exception>(out thrownException) == null)
                || (!Contract.Result<bool>() && Contract.ValueAtReturn<Exception>(out thrownException) != null));

            Contract.Assert(!m_disposed, "Attempting to use a disposed ActorClient");

            // We could build this atop CallMethodAsync<T>, but that would involve some extra continuations and
            // exception throwing, so I'll do a fresh implementation here.
            var seqNo = GetNewSequenceNumber();
            object[] args = new object[parameters.Length + 1];
            args[0] = methodName;
            Array.Copy(parameters, 0, args, 1, parameters.Length);
            var request = new ActorServiceRequest(ActorOperations.CALLMETHOD, myId, seqNo, args);
            var completionTask = SendActorRequestAsync(request);
            try
            {
                var response = completionTask.Result;
                if (response.Status == ActorServiceResultStatus.Succeeded)
                {
                    result = response.GetTypedResult<T>();
                    thrownException = null;
                    return true;
                }
                else
                {
                    result = default(T);
                    thrownException = ConstructException(response);
                    return false;
                }
            }
            catch (Exception e)
            {
                var baseException = (e is AggregateException) ? ((AggregateException)e).GetBaseException() : e;
                thrownException = baseException;
                result = default(T);
                return false;
            }
        }

        public object CallMethod(string methodName, object[] parameters)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);

            return CallMethod<object>(methodName, parameters);
        }

        public T CallMethod<T>(string methodName, object[] parameters)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);

            T result;
            Exception exception;
            if (!CallMethodNoThrow<T>(methodName, parameters, out result, out exception))
                throw exception;
            return result;
        }

        public void CallMethodFireAndForget(string methodName, object[] parameters)
        {
            Contract.Requires(parameters != null);

            Contract.Assume(!m_disposed, "Attempting to use a disposed ActorClient");  // Static checker can't prove this.  Still, check at runtime.

            var args = new object[parameters.Length + 1];
            args[0] = methodName;
            Array.Copy(parameters, 0, args, 1, parameters.Length);
            var request = new ActorServiceRequest(ActorOperations.CALLMETHOD, myId, ActorServiceRequest.FireAndForgetSequenceNumber, args);
            lock (m_operationLockObject)
            {
                Task t = m_actorTcpClient.TransmitRequestAsync(request);
                t.Wait();  // Make sure multiple requests don't stomp over each other.  Or do we do this internally instead?
            }
        }

        public Task<T> CallMethodAsync<T>(string methodName, object[] parameters)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);

            Contract.Assert(!m_disposed, "Attempting to use a disposed ActorClient");

            var seqNo = GetNewSequenceNumber();
            var args = new object[] { methodName }.Concat(parameters).ToArray();
            var request = new ActorServiceRequest(ActorOperations.CALLMETHOD, myId, seqNo, args);
            var completionTask = SendActorRequestAsync(request);
            return completionTask.ContinueWith(antecedent =>
                {
                    var result = antecedent.Result; // Will throw exception on communication failure
                    if (result.Status == ActorServiceResultStatus.Succeeded)
                    {
                        return result.GetTypedResult<T>();
                    }
                    else throw ConstructException(result); // Throws exception on operational failure
                });
        }

        // Yields a Task<string>, which yields JSON result string
        public Task<string> CallMethodAsyncRaw(string methodName, object[] parameters)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);

            Contract.Assert(!m_disposed, "Attempting to use a disposed ActorClient");

            var seqNo = GetNewSequenceNumber();
            var args = new object[] { methodName }.Concat(parameters).ToArray();
            var request = new ActorServiceRequest(ActorOperations.CALLMETHOD, myId, seqNo, args);
            var completionTask = SendActorRequestAsync(request);
            return completionTask.ContinueWith(antecedent =>
            {
                var result = antecedent.Result; // Will throw exception on communication failure
                if (result.Status == ActorServiceResultStatus.Succeeded)
                {
                    return result.Result;
                }
                else throw ConstructException(result); // Throws exception on operational failure
            });
        }

        public Task CallMethodAsyncVoid(string methodName, object[] parameters)
        {
            Contract.Requires(!String.IsNullOrEmpty(methodName));
            Contract.Requires(parameters != null);

            Task task = Task.Factory.StartNew(() =>
            {
                Object result;
                Exception exception;
                if (!CallMethodNoThrow<object>(methodName, parameters, out result, out exception))
                    throw exception;
                Contract.Assert(result == null, "While calling a method, we thought it should have returned null, but it returned an object.");
            });
            return task;
        }

        #endregion CallMethod variants

        #region Subscription methods
        // Subscribes to an eventType, both locally and on the actor
        public IDisposable Subscribe(string eventType, IObserver<ActorEvent> eventObserver)
        {
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !eventType.Contains(reservedChar)), "Actor event names cannot contain reserved delimiters like '$'.");

            Contract.Assert(!m_disposed, "Attempting to use a disposed ActorClient");

            return SubscribeAsync(eventType, eventObserver).Result;
        }

        private Task<IDisposable> SubscribeAsync(string eventType, IObserver<ActorEvent> eventObserver)
        {
            if (m_observers.ContainsKey(eventType))
            {
                throw new InvalidOperationException("Attempted multiple subscriptions to " + eventType);
            }

            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.SUBSCRIBE, myId, seqNo, eventType);
            // For replay-type events (like "ETW"), we need to have a handler in place before
            // we subscribe, because the act of subscribing may result in events being sent back.
            m_observers[eventType] = eventObserver;
            var completionTask = SendActorRequestAsync(asreq);
            return completionTask.ContinueWith(antecedent =>
                {
                    var asres = antecedent.Result; // will throw on network failure
                    if (asres.Status == ActorServiceResultStatus.Succeeded)
                    {
                        return (IDisposable)new DisposableWrapper(eventType, this);
                    }
                    else
                    {
                        m_observers.Remove(eventType);
                        throw ConstructException(asres); // will throw on operational failure
                    }
                }, TaskContinuationOptions.ExecuteSynchronously);
        }

        // Allow for explicit unsubscription from an event type, since IDisposable from original Subscribe() may no longer be available.
        internal void Unsubscribe(string eventType)
        {
            UnsubscribeAsync(eventType).Wait();
        }

        private Task UnsubscribeAsync(string eventType)
        {
            IObserver<ActorEvent> observer;
            if (m_observers.TryGetValue(eventType, out observer))
            {
                observer.OnCompleted();
                m_observers.Remove(eventType);
            }

            //Console.WriteLine("ActorClient Unsubscribe({0}): removed observer", eventType);
            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.UNSUBSCRIBE, myId, seqNo, eventType);
            var completionTask = SendActorRequestAsync(asreq);
            return completionTask.ContinueWith(antecedent =>
                {
                    var asres = antecedent.Result; // will throw on communication failure
                    if (asres.Status == ActorServiceResultStatus.Succeeded) return;
                    else throw ConstructException(asres);
                }, TaskContinuationOptions.ExecuteSynchronously);
        }

        // ActorStateObservable's Dispose needs to unsubscribe, but doesn't care about the results.
        // Additionally, we hit a deadlock that implied calling Unsubscribe and blocking for it to complete from
        // an actor method was a bad idea.
        public void UnsubscribeFireAndForget(string eventType)
        {
            IObserver<ActorEvent> observer;
            if (m_observers.TryGetValue(eventType, out observer))
            {
                observer.OnCompleted();
                m_observers.Remove(eventType);
            }

            //Console.WriteLine("ActorClient UnsubscribeFireAndForget({0}): removed observer", eventType);
            var seqNo = ActorServiceRequest.FireAndForgetSequenceNumber; // No response required
            var asreq = new ActorServiceRequest(ActorOperations.UNSUBSCRIBE, myId, seqNo, eventType);
            m_actorTcpClient.TransmitRequestAsync(asreq);
            // No need to wait for the result.
        }

        internal void UnsubscribeAll()
        {
            UnsubscribeAllAsync().Wait();
        }

        private Task UnsubscribeAllAsync()
        {
            foreach (var observer in m_observers.Values)
                observer.OnCompleted();
            m_observers.Clear();

            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.UNSUBSCRIBEALL, myId, seqNo);
            var completionTask = m_actorTcpClient.TransmitRequestAsync(asreq);
            return completionTask.ContinueWith(antecedent =>
                {
                    var asres = antecedent.Result;
                    if (asres.Status == ActorServiceResultStatus.Succeeded) return;
                    else throw (ConstructException(asres));
                }, TaskContinuationOptions.ExecuteSynchronously);
        }

        // Use this during Dispose() so that debugger doesn't break on result processing exception
        public void UnsubscribeAllFireAndForget()
        {
            foreach (var observer in m_observers.Values)
                observer.OnCompleted();
            m_observers.Clear();

            var seqNo = ActorServiceRequest.FireAndForgetSequenceNumber; // No response required
            var asreq = new ActorServiceRequest(ActorOperations.UNSUBSCRIBEALL, myId, seqNo);

            try
            {
                // Called during shutdown code paths, and could get into a race with logic trying to reconnect or dispose.
                var localActorTcpClient = m_actorTcpClient;
                if (localActorTcpClient != null)
                    localActorTcpClient.TransmitRequestAsync(asreq);
                // No need to wait for result.
            }
            catch (ObjectDisposedException)
            { }
            catch (InvalidOperationException)
            { }
        }

        #endregion Subscription methods

        #region Configuration

        // Configuration string format: Foo=x;Bar=y
        public void Configure(string configurationString)
        {
            var seqNo = GetNewSequenceNumber();
            ActorServiceRequest request = new ActorServiceRequest(ActorOperations.CONFIGURE, myId, seqNo, configurationString);
            Exception thrownException;
            var completionTask = SendActorRequestAsync(request);
            try
            {
                var response = completionTask.Result;
                if (response.Status == ActorServiceResultStatus.Succeeded)
                {
                    return;
                }
                else
                {
                    thrownException = ConstructException(response);
                }
            }
            catch (Exception e)
            {
                var baseException = (e is AggregateException) ? ((AggregateException)e).GetBaseException() : e;
                thrownException = baseException;
            }

            if (thrownException != null)
                throw thrownException;
        }

        #endregion Configuration

        #region Lifetime Management

        // We'll treat this as a best-effort attempt.  We want to make forward progress if this doesn't work.
        // We will not reconnect to the actor after this message has been sent.
        public Task SendDeleteRequest()
        {
            if (m_disposed) return Task.FromResult(true);

            var seqNo = GetNewSequenceNumber();
            var asreq = new ActorServiceRequest(ActorOperations.DELETEACTOR, myId, seqNo, m_partitionName);
            var completionTask = SendActorRequestAsync(asreq);
            m_sentDeleteRequest = true;
            return completionTask.ContinueWith(antecedent =>
            {
                // Clear the event handler.  We don't intend to reconnect anymore.
                m_actorTcpClient.Faulted -= new EventHandler(ActorClient_Faulted);

                var asres = antecedent.Result; // will throw on communication failure
                if (asres.Status == ActorServiceResultStatus.Succeeded)
                {
                    return asres.GetTypedResult<bool>();
                }
                else throw ConstructException(asres);
            }, TaskContinuationOptions.ExecuteSynchronously);
        }

        #endregion Lifetime Management

        //
        // Private/protected utility methods and classes
        //

        #region Request transmission methods
        // Construct an exception from an error string returned by an actor.
        internal static Exception ConstructException(ActorServiceResult asr)
        {
            Contract.Requires(asr.Status != ActorServiceResultStatus.Succeeded, "Called ConstructException for successful result");
            string exceptionType = asr.ExceptionType;
            string message = asr.ExceptionMessage;
            string stackTrace = asr.ExceptionStackTrace;

            return ConstructException(exceptionType, message, stackTrace);
        }

        public static Exception ConstructException(string exceptionType, string message, string origStackTrace = null)
        {
            Contract.Requires(!String.IsNullOrWhiteSpace(exceptionType));
            Contract.Requires(exceptionType.EndsWith("Exception"));

            Exception exception;
            Func<string, Exception> exceptionCreator;
            if (s_knownExceptionTable.TryGetValue(exceptionType, out exceptionCreator))
            {
                exception = exceptionCreator(message);
            }
            else
            {
                exception = new Exception(String.Format("{0}:{1}", exceptionType, message));
            }

            if (origStackTrace != null)
                exception.Data["OriginalStackTrace"] = origStackTrace;

            return exception;
        }

        private void LaunchActorRequestAttempt(ActorServiceRequest request,
                               TaskCompletionSource<ActorServiceResult> tcs,
                               int retriesLeft)
        {
            if (m_disposed)
                throw new ObjectDisposedException("ActorClient");

            Task<ActorServiceResult> completionTask = null;
            // This code can race with the reconnection logic, both taking the same lock.  We might see a null m_actorTcpClient,
            // and if we give up the lock and try again, we might see a valid one again.  Maybe.
            int numRetries = 5;
            while (numRetries-- >= 0 && completionTask == null)
            {
                lock (m_operationLockObject)
                {
                    if (m_disposed)
                        throw new ObjectDisposedException("ActorClient");

                    // Lock out operations if channel recovery operations are occurring.
                    if (m_actorTcpClient != null)
                        completionTask = m_actorTcpClient.TransmitRequestAsync(request);
                }
                if (completionTask == null)
                    Task.Delay(1000);  // Give the reconnection logic a chance to run.
            }

            if (completionTask == null)
            {
                //Contract.Assert(m_actorTcpClient != null, "We reconnected, but TransmitRequestAsync returned a null task.  Subtle race condition?");
                throw new ObjectDisposedException("ActorClient");
            }

            completionTask.ContinueWith(antecedent =>
            {
                if (antecedent.IsFaulted)
                {
                    if (retriesLeft > 0)
                    {
                        // Retry it.
                        LaunchActorRequestAttempt(request, tcs, retriesLeft - 1);
                    }
                    else
                    {
                        // Kill it
                        tcs.TrySetException(antecedent.Exception.GetBaseException());
                    }
                }
                else
                {
                    var result = antecedent.Result;
                    switch (result.Status)
                    {
                        // Filter out and handle special-case error modes

                        case ActorServiceResultStatus.TryAgain:
                            //Console.WriteLine("ActorClient: Responding to TryAgain/Transient error");
                            // We have a transient problem; perhaps a quorum is not yet available.
                            // Wait some time, then re-try the attempt without downcounting.
                            Task.Delay(2000).ContinueWith(_ =>
                            {
                                LaunchActorRequestAttempt(request, tcs, retriesLeft);
                            }, TaskContinuationOptions.ExecuteSynchronously);
                            break;

                        case ActorServiceResultStatus.WrongTarget:
                            //Console.WriteLine("ActorClient: Responding to WrongTarget/Not Primary error, retries left: {1}",
                            //    retriesLeft);
                            // We're connected to a replica that is no longer the primary.
                            // We need to fault the channel and retry the command.
                            if (retriesLeft > 0)
                            {
                                // Give this a nice long delay.  Fabric is confused.
                                Task.Delay(5000).ContinueWith(async _ =>
                                {
                                    //m_resolvedServicePartition = null; // try nulling this out?
                                    await RecoverFromDisconnectAsync().ConfigureAwait(continueOnCapturedContext:false);
                                    LaunchActorRequestAttempt(request, tcs, retriesLeft - 1);
                                }, TaskContinuationOptions.ExecuteSynchronously);
                            }
                            else tcs.TrySetResult(result);
                            break;

                        default:
                            // This is either an operational exception or a success.  Pass it on to the caller.
                            tcs.TrySetResult(result);
                            break;
                    }
                }
            });
        }

        private Task<ActorServiceResult> SendActorRequestAsync(ActorServiceRequest request)
        {
            TaskCompletionSource<ActorServiceResult> tcs = new TaskCompletionSource<ActorServiceResult>();
            LaunchActorRequestAttempt(request, tcs, COMMAND_RETRIES);

            return tcs.Task;
        }

        #endregion Request transmission methods

        #region Channel fault recovery
        // Algorithm to recover from channel fault.
        protected async Task<bool> RecoverFromDisconnectAsync()
        {
            if (m_disposed) return false;

            var myHashCode = this.GetHashCode();
            var message = String.Format("{0}: ActorClient({1}), actor {2}, entered RecoverFromDisconnect()", DateTime.Now, myHashCode, m_actorUri);
            Trace.WriteLine(message, "Error");
            //Console.WriteLine(message);

            // Attempt to reestablish a connection.
            int MAX_BACKOFF_MS = 600000; // = 10 minutes
            int backoff = 1000; // one second, initially
            bool connected = false;
            bool lastTry = false;
            while (!connected)
            {
                if (m_disposed) return false;
                // Dispose of the current channel
                m_actorTcpClient.Dispose();

                try
                {
                    bool succeeded = await SetupActorClientAsync(false).ConfigureAwait(continueOnCapturedContext:false);
                    if (!succeeded)
                        return false;  // IE, we looked up the service and it no longer exists.
                    Trace.WriteLine(String.Format("ActorClient({0}) successfully re-established connection", myHashCode), "Error");
                    connected = true;
                }
                catch
                {
                }

                if (!connected)
                {
                    // Give up after a certain number of attempts.  Don't want many attempts to reconnect for days.
                    if (lastTry)
                        return false;

                    Trace.WriteLine(String.Format("ActorClient({0}) Backing off {1} ms before re-trying connection", myHashCode, backoff), "Error");
                    await Task.Delay(backoff).ConfigureAwait(continueOnCapturedContext:false);

                    if (backoff == MAX_BACKOFF_MS)
                        lastTry = true;

                    if (backoff < MAX_BACKOFF_MS)
                    {
                        backoff = backoff << 1;
                        if (backoff > MAX_BACKOFF_MS) backoff = MAX_BACKOFF_MS;
                    }
                }
            }

            try
            {
                if (m_observers.Count > 0)
                {
                    foreach (var kvp in m_observers)
                    {
                        if (m_disposed) return false;

                        // Resubscribe to everything
                        var seqNo = GetNewSequenceNumber();
                        var req = new ActorServiceRequest(ActorOperations.SUBSCRIBE, myId, seqNo, kvp.Key);
                        var name = (m_partitionName == null) ? m_actorUri.ToString() : m_actorUri.ToString() + ", " + m_partitionName;
                        var msg = String.Format("**AC.RecoverFromDisconnectAsync():Resubscribing to {0}, event {1}",
                            name, kvp.Key);
                        Trace.WriteLine(msg, "Info");
                        Console.WriteLine(msg);
                            
                        var localCopyOfActorTcpClient = m_actorTcpClient;
                        if (m_disposed || localCopyOfActorTcpClient == null) return false;

                        await localCopyOfActorTcpClient.TransmitRequestAsync(req).ConfigureAwait(continueOnCapturedContext: false);
                    }
                }
            }
            catch (Exception exc)
            {
                // We failed to resubscribe.  Return false so that we can reconnect again.
                var baseException = (exc is AggregateException) ? ((AggregateException)exc).GetBaseException() : exc;
                var msg = String.Format("RecoverFromDisconnect({0}): Exception during resubscription: {1}",
                    m_actorUri.ToString(), baseException.Message);
                Trace.WriteLine(msg, "Error");
                Console.WriteLine(msg);
                return false;
            }


            // OK, we're connected.  Re-register the fault handler.
            message = String.Format("{0}: ActorClient({1}): RecoverFromDisconnect completed successfully", DateTime.Now, m_actorUri);
            Trace.WriteLine(message, "Error");
            Console.WriteLine(message);
            m_actorTcpClient.Faulted += new EventHandler(ActorClient_Faulted);
            return true;
        }

        // Fault handler for underlying channel.
        protected void ActorClient_Faulted(object sender, EventArgs e)
        {
            TimeSpan maxRetryTime = TimeSpan.FromMinutes(10);
            lock (m_operationLockObject) // lock out operations while we're recovering from fault
            {
                if (m_actorTcpClient == null || m_disposed)
                    return;  // Nothing to do, and not harmful.  Not worth logging.

                if (m_actorTcpClient != null && m_actorTcpClient.IsShuttingDown)
                    return;

                if (m_actorTcpClient != null && m_actorTcpClient.IsFaulted)
                {
                    m_resolveState = ResolverState.Resolving;

                    var message = String.Format("ActorClient({0}{1}) detected faulted channel, entering disconnect recovery logic.",
                        m_actorUri,
                        (m_partitionName == null) ? String.Empty : ", " + m_partitionName);
                    Trace.WriteLine(message, "Error");
                    Console.WriteLine(message);

                    var startTime = DateTimeOffset.UtcNow;
                    var recoverTask = RecoverFromDisconnectAsync();
                    // Note: We're synchronously blocking here, but we kinda need to know whether we succeeded.
                    // Consider using a continuation in the future, perhaps?
                    while (!recoverTask.Result && !m_disposed && (m_actorTcpClient != null && !m_actorTcpClient.IsShuttingDown))
                    {
                        Task.Delay(1000).Wait();
                        var currentTime = DateTimeOffset.UtcNow;
                        if (currentTime - startTime > maxRetryTime)
                        {
                            message = String.Format("ActorClient({0}{1}) giving up on reconnecting to a faulted channel.",
                                m_actorUri,
                                (m_partitionName == null) ? String.Empty : ", " + m_partitionName);
                            Console.WriteLine(message);
                            Trace.WriteLine(message, "Error");
                            return;
                        }
                        recoverTask = RecoverFromDisconnectAsync();  // Try again.
                    }
                }
                else
                {
                    Trace.WriteLine(String.Format("Fault handler({0}): not faulted, no action taken", m_actorUri), "Info");
                }
            }
        }

        #endregion Channel fault recovery

        #region Channel setup

        // The assuming this method's various implementations require platform-specific networking support.
        protected abstract Task<bool> SetupActorClientAsync(bool registerFaultedCallback = true);

        // Ensures that we can send a command, AND that
        // we've connected to the correct actor.  (If we resolved the service too soon after
        // a line went dead, we may have resolved to a stale address.)  The Ping() actor service method
        // will return the actor's service name (e.g., "fabric:/actor/list/mylist").
        protected async Task<bool> TestConnectivityAsync(ActorRawTcpClient candidate)
        {
            Contract.Requires(candidate != null);

            var intendedActor = m_actorUri.ToString();
            for (int j = 0; j < COMMAND_RETRIES; j++)
            {
                try
                {
                    var seqNo = GetNewSequenceNumber();
                    var pingRequest = new ActorServiceRequest(ActorOperations.PING, myId, seqNo);
                    var pingTask = candidate.TransmitRequestAsync(pingRequest);
                    // In case the network connection is flaky or server crashes, time out and try again.
                    // Using a progressive backoff approach here.
                    int timeout = Math.Max((j + 1) * TestConnectionTimeoutMS / COMMAND_RETRIES, MinConnectionTimeoutMS);
#if DEBUG
                    if (Debugger.IsAttached)
                        timeout = Math.Min(timeout, MinDebuggerAttachedConnectionTimeoutMS);
                    //timeout = Timeout.Infinite;
#endif
                    if (pingTask != await Task.WhenAny(pingTask, Task.Delay(timeout)).ConfigureAwait(continueOnCapturedContext: false))
                    {
                        Console.WriteLine("ActorClient.TestConnectivity timed out after {0} seconds!", timeout / 1000);
                        continue;
                    }
                    var pingResponse = await pingTask.ConfigureAwait(continueOnCapturedContext:false);
                    if (pingResponse.Status == ActorServiceResultStatus.Succeeded)
                    {
                        var connectedActor = pingResponse.GetTypedResult<string>();
                        if (connectedActor.Equals(intendedActor))
                        {
                            return true;
                        }
                        else
                        {
                            Console.WriteLine("TestConnectivity: Connected to wrong actor ({0},{1})!  Status: {2}  Re-resolving.",
                                intendedActor,
                                connectedActor,
                                pingResponse.Status);
                            return false;
                        }
                    }
                }
                catch
                {
                    // Socket disconnected?
                    if (candidate.IsFaulted) return false;
                }

                await Task.Delay(1000).ConfigureAwait(continueOnCapturedContext: false);
            }

            return false;
        }

        private Uri MakeGatewayAddress(Uri fabricUri)
        {
            // Convert fabricUri to gatewayUri
            var gatewayString = fabricUri.ToString();
            var useLocalGateway = fabricUri.Host.Contains("localhost") || fabricUri.Host.Contains("127.0.0.1") || fabricUri.Host.Contains("[::1]");
            var portString = "" + fabricUri.Port;
            gatewayString = gatewayString.Replace(portString, useLocalGateway ? "8506" : "8505");
            return new Uri(gatewayString);
        }

        #endregion Channel setup

        // Utility class that wraps unsubscription information in an IDisposable
        class DisposableWrapper : IDisposable
        {
            private readonly string m_eventType;
            private readonly ActorClient m_actorClient;

            public DisposableWrapper(string eventType, ActorClient actorClient)
            {
                m_eventType = eventType;
                m_actorClient = actorClient;
            }

            public void Dispose()
            {
                try
                {
                    m_actorClient.Unsubscribe(m_eventType);
                    //Console.WriteLine("ActorClient DW.Dispose(): Unsubscribed {0} from actor", m_eventType);
                }
                catch(Exception)
                {
                    // And ignore any problems with the unsubscribe
                }
            }
        }


        #region IObserver<ActorEvent> implementation

        public void OnCompleted()
        {
            if (m_observers.Count > 0)
            {
                Contract.Assert(false, "ActorClient::OnCompleted called, and we had at least 1 observer.  Should we close them all?");
                foreach (var observer in m_observers.Values)
                    observer.OnCompleted();
                m_observers.Clear();
            }

            Dispose();
        }

        public void OnError(Exception error)
        {
            throw new NotImplementedException();
        }

        public void OnNext(ActorEvent value)
        {
            var eventType = value.EventType;
            IObserver<ActorEvent> observer;

            // Forward event to any interested observer
            if (m_observers.TryGetValue(eventType, out observer))
            {
                observer.OnNext(value);
            }

            // Support wildcarding
            if (m_observers.TryGetValue("*", out observer))
            {
                observer.OnNext(value);
            }
        }

        #endregion IObserver<ActorEvent> implementation
    }
}
